<?php
/**
 * 抓取控制器
 * @author lileino9
 *
 */
namespace app\crontab\controller\spider;
use app\common\service\PidFile;
use app\crontab\controller\CrontabCommon;

class Spider extends CrontabCommon {
	
	/**
	 * 启动抓取采集源列表子进程个数
	 * @var integer
	 */
	const runSpiderSourceUrlProcess = 5;
	
	/**
	 * 启动采集下载详情页子进程个数
	 * @var integer
	 */
	const runSpiderDownLoadProcess = 10;
	
	/**
	 * 列表采集列表集合延时时间
	 * @var integer
	 */
	const SpiderSourceUrlDelaySecond = 0;
	
	/**
	 * 初始化方法
	 */
	protected function _initialize()
	{
		//先执行父类初始化
		parent::_initialize();
	}
	
	/**
	 * "空" 方法
	 * @param string $method 方法名
	 * @param array $args 参数
	 */
	public function _empty( $method, $args ) {
		$sub_action = $_GET['sub_action'];
		if ( $sub_action != "spider_source_url" ) {
			return false;
		}
 		$model = $method;
 		// 进程控制
 		if (!PidFile::createPidFile($sub_action."_".$model, 1)) {
 			echo '进程已启动足够数量' . PHP_EOL;
 			return false;
 		}
 		
 		if ( !class_exists($model) ) {
 			echo "Model[{$model}]不存在".PHP_EOL;
 			return false;
 		}
 		// 验证模型方法
 		if ( !method_exists($model, SpiderSourceModel::runSpiderList) ) {
 			echo "Model[{$model}]没有列表采集入口方法".SpiderSourceModel::runSpiderList;
 			return false;
 		}
 		if ( !method_exists( $model , SpiderSourceModel::runSpiderDetail) ) {
 			echo "Model[$model]没有详情采集入口方法".SpiderSourceModel::runSpiderDetail.PHP_EOL;
 			return false;
 		}
 		
 		// 资源编号
 		$source_id = $_GET['source_id'];
 		// 采集源采集列表采集
 		$source_url_map = [
 				'source_id' => $source_id,
 				'status' => 1,
 		];
 		$modelSpiderUrl = new SpiderUrlModel();
 		$urlList = $modelSpiderUrl->where($source_url_map)->select();
 		if ( !$urlList ) {
 			echo '没有采集资源可用' . PHP_EOL;
 			return;
 		}
 		
 		// 计算总数量
 		$total = count($urlList);
 		// 子进程个数
 		$process = self::runSpiderSourceUrlProcess;
 		// 每个进程数据数量
 		$limit = ceil($total/$process);
 		// 数据切割分组
 		$all_data_chunk = array_chunk($urlList, $limit);
 		// 主进程池
 		$pid_list = array();
  		
 		// 多进程处理
 		foreach ( $all_data_chunk as $kp=>$vp ) {
 			$pid = pcntl_fork();
 			if( $pid==-1 ){
 				die("can not fork");
 			}elseif ( $pid ) {
 				$pid_list[] = $pid;
 			}else {
 				// 执行代码
 				$modelSpiderSource = new SpiderSourceModel();
 				$modelSpiderUrl = new SpiderUrlModel();
 				$modelSpiderDownLoad = new SpiderDownloadModel();
 				$downLoadList = [];
 				$updateSpiderUrl = [];
 				foreach ( $vp as $data ) {
 					$url_type = $data['url_type'];
 					$target_url = $data['url'];
 					$modelObject  = new $data['model']();
 					$res = [];
 					if ( $url_type==SpiderUrlModel::urlListType ) {
 						// 列表
 						$method = SpiderSourceModel::runSpiderList;
 						$res = call_user_func_array ( array (
 								&$modelObject,
 								$method
 						), array($target_url) );
 					}elseif ( $url_type==SpiderUrlModel::urlDetailType ) {
 						// 详情直接放入结果集
 						$res[] = $target_url;
 					}
 					if ( $res  ) {
 						foreach ( $res as $rUrl ) {
 							$property = [];
 							if ( is_array($rUrl) ) {
 								$url = $rUrl['url'];
 								$property = $rUrl;
 							}elseif ( is_string($rUrl) ) {
 								$url = $rUrl;
 							}
 							// 判断Url是否存在
 							$urlRes = $modelSpiderDownLoad->get_data_by_unique_key('url', $url);
 							if ( $urlRes ) {
 								echo "url下载地址存在".PHP_EOL;
 								echo $url.PHP_EOL;
 								echo $modelSpiderDownLoad->getLastSql().PHP_EOL;
 								continue;
 							}
 							$downLoadList[] = array(
 									'parent_id' => $data['id'],
 									'source_id' => $data['source_id'],
 									'model' => $data['model'],
 									'url' => $url,
 									'property' => CommonFun::json_encode2($property),
 							);
 						}
 					}
 					// 更新采集资源表
 					$updateSpiderUrl[] = $data['id'];
 				}
 				// 处理采集结果集
 				if ( $downLoadList ) {
 					foreach ( $downLoadList as $insert ) {
 						$add_res = $modelSpiderDownLoad->add_data($insert);
 						if ( !$add_res ) {
 							echo $modelSpiderDownLoad->getError().PHP_EOL;
 						}else {
 							// 添加成功 加入下载队列
 							$spiderRedis = new SpiderArticleRedis();
 							$spiderRedis->pushSpiderDownLoadQueue($add_res);
 						}
 					}
 				}
 				if ( $updateSpiderUrl ) {
 					$updateSpiderUrl = array_unique($updateSpiderUrl);
 					foreach ( $updateSpiderUrl as $pk ) {
 						$modelSpiderUrl->where( array('id'=>$pk) )->setInc("run_count");
 					}
 					$modelSpiderUrl->muti_get_data_by_id_list($updateSpiderUrl, false, false);
 				}
 				exit();
 			}
 		}// End chunk foreach
 		while(count($pid_list) > 0)
 		{
 			$myId = pcntl_waitpid(-1, $status, WNOHANG);
 			foreach($pid_list as $key => $pid)
 			{
 				if($myId == $pid) unset($pid_list[$key]);
 			}
 		}
	}
	
	/**
	 * 执行采集程序
	 */
	public function run_spider_source_url() {
		//读取配置使用对应   php脚本  或  shell脚本模版
		CommonFun::get_other_config('CronConfig',2);
		// 获取任务
		$modelSourceModel = new SpiderSourceModel();
		$source_map = array(
				'status' => 1,
				'deleted' => -1,
		);
		$source_data_list = $modelSourceModel->muti_get_data_by_where($source_map);
		$current_mission_list = [];
		if ( $source_data_list ) {
			foreach ( $source_data_list as $v ) {
				$start = $v['start'];
				if ( $start==1 ) {
					$current_mission_list[] = $v;
				}
			}
		}
		// PHP 执行路径
		$PHP_EXE_PATH = C('MODE_PHP');
		if ( !$current_mission_list ) {
			// 没有任务要执行
			echo "No Missions".PHP_EOL;
		}else {
			$intNum = count($current_mission_list);
			// 进程PID数组
			$pids = array();
			
			// 创建任务
			for($i = 0; $i < $intNum; $i++) {
				// 产生子进程，而且从当前行之下开试运行代码，而且不继承父进程的数据信息
				$pids[$i] = pcntl_fork();
				if ($pids[$i] == -1) {
					//错误处理：创建子进程失败时返回-1.
					die('could not fork');
				}elseif ($pids[$i]) {
					//父进程会得到子进程号，所以这里是父进程执行的逻辑
					//pcntl_wait($status); //等待子进程中断，防止子进程成为僵尸进程。
				}else {
					// 开始时间
					$start_time = time();
					// 命令
					$go_command= "";
					// 任务信息
					$mission_info = $current_mission_list[$i];
					// 任务脚本标识->会直接跳入_empty方法
					$cron_script = "/".MODULE_NAME."/".$mission_info['model']."/sub_action/spider_source_url/source_id/".$mission_info['id'];
					// 生成命令
					$go_command = $PHP_EXE_PATH . ' ' . $cron_script . ' ' . ENV . ' ' . '127.0.0.1';
					// 执行命令
 					$a = exec($go_command, $out, $status);
 					// 运行结果没有成功
 					$last_result = 1;
 					if(!strstr(array_pop($out),'success') || strstr(array_pop($out),'failure') || $status>0 ){
 						Log::record("\r\n{$cron_script} 执行结果:\r\nOut:".var_export($out, true)."\r\nStatus:".var_export($status, true), Log::INFO);
 						$last_result = 2;
 					}
 					
 					// 更新数据库
 					$end_time = time();
 					$cost_time = intval($end_time - $start_time);
 					$cron_counts = $mission_info['cron_counts']+1;
 					$cron_id = $mission_info['id'];
 					$update = array(
 							'cron_counts' => $cron_counts,
 							'cost_time' => $cost_time,
 							'last_result'=>$last_result,
 							'last_run_time' => $end_time,
 					);
 					$CrontabModel = new SpiderSourceModel();
 					$CrontabModel->update_data($cron_id, $update);
					exit();
				}
				
			}
			
			while(count($pids) > 0)
			{
				$myId = pcntl_waitpid(-1, $status, WNOHANG);
				foreach($pids as $key => $pid)
				{
					if($myId == $pid) unset($pids[$key]);
				}
			}
			
		}
		
		// 输出执行的计划任务
		$exec_cron = $current_mission_list ? "\r\n".implode("\r\n", array_column($current_mission_list, "name")) : "None";
		echo "exec cron: {$exec_cron}".PHP_EOL;
		Log::record("Crontab Run At ".date("Y-m-d H:i:s"), Log::INFO);
	}
	
	/**
	 * 执行下载队列
	 */
	public function run_spider_down_load_queue() {
		// 进程个数控制
		if (!PidFile::createPidFile('run_spider_down_load_queue', self::runSpiderDownLoadProcess)) {
			echo '进程已启动足够数量' . PHP_EOL;
			return;
		}
		// 获取队列信息
		$queueModel = new SpiderArticleRedis();
		// 一个进程执行的次数
		$frequency_count = 300;
		// 空队列休息时间
		$queue_empty_sleep = 1;
		// 初始化计数
		$init_count = 1;
		// 下载模型
		$modelSpiderDownLoad = new SpiderDownloadModel();
		
		// 常驻队列
		while ( $init_count<=$frequency_count ) {
			// 对次数进行累加
			$init_count++;
			// 当前队列数量
			$queueSize = $queueModel->sizeSpiderDownLoadQueue();
			// 空队列处理
			if ( !$queueSize ) {
				sleep($queue_empty_sleep);
				continue;
			}
			// 取出一条
			$queueData = $queueModel->popSpiderDownLoadQueue();
			// 没有数据, 防止并发
			if ( !$queueData ) {
				continue;
			}
			//获取下载数据
			$downData = $modelSpiderDownLoad->get_data_by_id($queueData);
			if ( !$downData || $downData['is_down']==1 ) {
				// 无数据或者已下载
				continue;
			}
			
			// 采集下载主键
			$pk = $downData['id'];
			// 采集模型
			$spiderModel = $downData['model'];
			// 下载地址
			$target_url = $downData['url'];
			
			// 执行单条详情采集
			$obj  = new $spiderModel();
			$method = SpiderSourceModel::runSpiderDetail;
			$res = call_user_func_array ( array (
					$obj,
					$method
			), array($target_url) );
			// 更新结果
			$update = array(
					'run_count' => ++$downData['run_count'],
					'is_success' => $res===false ? -1 : 1,
					'is_down' => -1,
					'publish_id' => 0,
			);
			
			// 执行发布采集
			if ( $res ) {
				$publishMethod = SpiderSourceModel::publishSpiderData;
				$publish_res = call_user_func_array ( array (
						$obj,
						$publishMethod
				), array($res) );
				// 发布失败
				if ( !$publish_res ) {
					$error = $obj->getError();
					echo "编号:{$pk}采集失败,发布失败[{$error}]".PHP_EOL;
					$update['is_success'] = -1;
				}else {
					// 发布成功
					$update['is_publish'] = 1;
					$update['publish_id'] = is_array($publish_res) ? $publish_res['id'] : (int)$publish_res;
					$update['publish_model'] = $publish_res['publish_model'];
					$update['is_down'] = 1;
                    SpiderArticleJobs::spiderArticleTagPush( $update['publish_id'], SpiderArticleJobs::CONTENT_TYPE_ARTICLE );
				}
			}else {
				echo "编号:{$pk}采集失败,采集结果为空[".$obj->getError()."]".PHP_EOL;
				$update['is_success'] = -1;
			}
			
			// 更新采集下载数据
			$update_res = $modelSpiderDownLoad->update_data($pk, $update);
			if ( !$update_res ) {
				$error = $modelSpiderDownLoad->getError();
				echo "编号:{$pk}更新采集下载数据失败[{$error}]".PHP_EOL;
			}
			
			//对次数进行累加
			//$init_count++;
		}		
		
	}

	
	/**
	 * 重置丢失采集
	 */
	public function spider_miss_reset() {
		$modelSpiderDown = new SpiderDownloadModel();
		// 查询条件
		$where = [
				'run_count' => 0,
				'is_success' => -1,
				'is_publish' => -1,
				'is_down' => -1,
		];
		$id_list = $modelSpiderDown->search_id_list_no_page($where);
		if ( $id_list && is_array($id_list) ) {
			// 添加成功 加入下载队列
			$spiderRedis = new SpiderArticleRedis();
			$spiderRedis->pushSpiderDownLoadQueue($id_list);
		}
	}
	
	/**
	 * 重置失败的采集
	 */
	public function spider_down_reset() {
		$modelSpiderDown = new SpiderDownloadModel();
		// 查询条件
		$where = [
				'is_success' => -1,
				'is_down' => -1,
				'run_count' => array('between', array(1,3)),
				'is_publish' => -1,
		];
		$id_list = $modelSpiderDown->search_id_list_no_page($where);
		var_export($id_list);
		if ( $id_list && is_array($id_list) ) {
			// 添加成功 加入下载队列
			$spiderRedis = new SpiderArticleRedis();
			$spiderRedis->pushSpiderDownLoadQueue($id_list);
		}
	}
	
	/**
	 * 刷新IP代理
	 */
	public function refresh_ip_proxy(){
		$modelIpProxy = new IpProxyModel();
		// 刷新配置
		$refresh_config = array(
				// 国内 https  05分钟至25分钟
				array(
						'protocol' =>IpProxyModel::ProtocolHttps,
						'num' => 10,
						'area' => IpProxyModel::AreaInSide,
						'time' => 1,
				),
				// 国内 https  26分钟至03小时
				array(
						'protocol' =>IpProxyModel::ProtocolHttps,
						'num' => 5,
						'area' => IpProxyModel::AreaInSide,
						'time' => 2,
				),
				// 国内 https 12小时至24小时
				array(
						'protocol' =>IpProxyModel::ProtocolHttps,
						'num' => 1,
						'area' => IpProxyModel::AreaInSide,
						'time' => 5,
				),
				
				// 国内 http  05分钟至25分钟
				array(
						'protocol' =>IpProxyModel::ProtocolHttp,
						'num' => 2,
						'area' => IpProxyModel::AreaInSide,
						'time' => 1,
				),
				// 国内 http  26分钟至03小时
				array(
						'protocol' =>IpProxyModel::ProtocolHttp,
						'num' => 5,
						'area' => IpProxyModel::AreaInSide,
						'time' => 2,
				),
				
				// 国内 http 12小时至24小时
				array(
						'protocol' =>IpProxyModel::ProtocolHttp,
						'num' => 1,
						'area' => IpProxyModel::AreaInSide,
						'time' => 5,
				),
				
		);
		
		// 获取IP
		foreach ( $refresh_config as $v ) {
			// 获取数量
			$num = $v['num'];
			// 稳定时长
			$time = $v['time'];
			// 区域
			$area = $v['area'];
			// 获取协议
			$protocol = $v['protocol'];
			// 检测是否还有可用IP
			$https_ip = $modelIpProxy->getIPproxy($protocol, $num, $area, false);
			if ( $https_ip===false ) {
				echo $modelIpProxy->getError().PHP_EOL;
				continue;
			}
			$count = count($https_ip);
			if ( $count<$num ) {
				$diff = $num-$count;
				// 获取ip地址并入库
				$res = $modelIpProxy->addNewIPproxy($diff, $protocol, $time, $area);
				if ( $res===false ) {
					echo $modelIpProxy->getError().PHP_EOL;
				}
				sleep(3);
			}
			
		}
	}
	
	/**
	 * 抓取coinfiy
	 */
	public function spider_coinfyi() {
		$spiderModel = SpiderCoinFyiModel::getInstance();
		$spiderModel->run();
	}
	
	/**
	 * 抓取详情数据
	 */
	public function spider_detail() {
	    $spiderModel = SpiderDetailModel::getInstance();
	    $spiderModel->run();
	}
	
	/**
	 * 抓取reddit
	 */
	public function spider_reddit() {
	    $spiderModel = SpiderRedditModel::getInstance();
	    $spiderModel->run();
	}
	
	/**
	 * 抓取steemit
	 */
	public function spider_steemit() {
	    $spiderModel = SpiderSteemitModel::getInstance();
	    $spiderModel->run();
	}
	
	
	
	
	
	
	
}
?>